package cn.demoncat.util.zookeeper;

import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import cn.demoncat.util.exception.ConnRuntimeException;
import cn.demoncat.util.zookeeper.config.ZookeeperProperties;
import cn.demoncat.util.zookeeper.constant.RetryType;

/**
 * Zookeeper 客户端连接工具
 * 
 * @author 延晓磊
 *
 * @since 2018年11月2日
 */
@Component
public class ZkConnectionUtil {

	// 配置
	private final ZookeeperProperties zookeeperProperties;
	// 连接池
	private final GenericObjectPool<CuratorFramework> pool;
 
	private static final Logger LOG = LoggerFactory.getLogger(ZkConnectionUtil.class);
	
	/**
	 * 初始化
	 */
	@Autowired
	public ZkConnectionUtil(ZookeeperProperties zookeeperProperties) {
		super();
		// 保存配置
		this.zookeeperProperties = zookeeperProperties;
		// 连接池配置
		GenericObjectPoolConfig<CuratorFramework> conf = new GenericObjectPoolConfig<>();
		conf.setMinIdle(zookeeperProperties.getPoolMinIdle());
		conf.setMaxIdle(zookeeperProperties.getPoolMaxIdle());
		conf.setMaxTotal(zookeeperProperties.getPoolMaxTotal());
		conf.setTestOnBorrow(zookeeperProperties.getPoolTestOnBorrow());	
		conf.setTimeBetweenEvictionRunsMillis(zookeeperProperties.getPoolCheckTime());
		// 连接对象工厂
		ZkPoolFactory factory = new ZkPoolFactory(zookeeperProperties.getUrl(), Integer.MAX_VALUE, zookeeperProperties.getConnectionTimeout());
		// 连接池
		pool = new GenericObjectPool<>(factory, conf);
	}

	/**
	 * 获取连接池
	 * 
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public GenericObjectPool<CuratorFramework> getPool(){
		// 获取数  pool.getBorrowedCount()
		// 返回数  pool.getReturnedCount()
		// 创建数  pool.getCreatedCount()
		// 销毁数  pool.getDestroyedCount()
		return pool;
	}
	
	/**
	 * 通过连接池获取ZK连接
	 * 
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public CuratorFramework getPoolConnection() {
		try {
			return pool.borrowObject();
		} catch (Exception e) {
			throw new ConnRuntimeException("通过连接池获取Zookeeper连接失败", e);
		}
	}
 
	/**
	 * 返回ZK连接到连接池（超过maxTotal仍未返回，再次获取时将阻塞等待）
	 * 
	 * @param conn
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public void returnPoolConnection(CuratorFramework conn) {
		pool.returnObject(conn);
	}
	
	/**
	 * 销毁连接池中获取的ZK连接(通常按照空闲原则自动销毁，不建议主动调用)
	 * 
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public void closePoolConnection(CuratorFramework conn) {
		try {
			pool.invalidateObject(conn);
		} catch (Exception e) {
			LOG.error("通过连接池销毁Zookeeper连接失败：conn = {}", conn);
		}
	}
	
	/**
	 * 直接获取ZK连接
	 * 
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public CuratorFramework getConnection() {
		return getConnection(zookeeperProperties.getSessionTimeout());
	}
	
	/**
	 * 直接获取ZK连接
	 * 
	 * @param sessionTimeout	会话超时毫秒
	 * 
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public CuratorFramework getConnection(int sessionTimeout) {
		// 使用工厂创建ZK客户端：ZK连接地址，会话超时，连接超时，重试机制
		CuratorFramework conn = CuratorFrameworkFactory.newClient(zookeeperProperties.getUrl(), sessionTimeout, zookeeperProperties.getConnectionTimeout(), RetryType.RNT);
		// 开始连接
		conn.start();
		LOG.info("开启Zookeeper连接：url = {}", zookeeperProperties.getUrl());
		return conn;
	}
	
	/**
	 * 直接关闭ZK连接
	 * 
	 * @return
	 * 
	 * @author 延晓磊
	 *
	 * @since 2018年11月2日
	 */
	public void closeConnection(CuratorFramework conn) {
		if (conn != null) {
			conn.close();
		}
	}
}
